{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "d238bcf6",
   "metadata": {},
   "outputs": [],
   "source": [
    "import findspark\n",
    "findspark.init()\n",
    "from pyspark import SparkContext, SparkConf\n",
    "conf = SparkConf()\n",
    "sc = SparkContext(conf=conf)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "e1619303",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 创建RDD parallelize\n",
    "rdd = sc.parallelize(['hello spark','hello python','hello pyspark'])\n",
    "rdd1 = rdd.flatMap(lambda x:x.split(\" \")).map(lambda x: (x,1))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "fd2793f5",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('hello', 1), ('spark', 1), ('hello', 1), ('python', 1), ('hello', 1), ('pyspark', 1)]\n"
     ]
    }
   ],
   "source": [
    "print(rdd1.collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "010a4df4",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 二元组相关算子， key-value型RDD"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "b26e150d",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('hello', 2), ('spark', 2), ('hello', 2), ('python', 2), ('hello', 2), ('pyspark', 2)]\n"
     ]
    }
   ],
   "source": [
    "# mapValues 进队value操作\n",
    "rdd2 = rdd1.mapValues(lambda x: x+1)\n",
    "print(rdd2.collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "79490694",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('python', [1]), ('spark', [1]), ('pyspark', [1]), ('hello', [1, 1, 1])]\n"
     ]
    }
   ],
   "source": [
    "# groupByKey\n",
    "rdd3 = rdd1.groupByKey()\n",
    "print(rdd3.map(lambda x: (x[0], list(x[1]))).collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "85f576ee",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('python', 1), ('spark', 1), ('pyspark', 1), ('hello', 3)]\n"
     ]
    }
   ],
   "source": [
    "# reduceByKey\n",
    "rdd4 = rdd1.reduceByKey(lambda x,y: x+y)\n",
    "print(rdd4.collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "ab31bc77",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('b', (1, 0)), ('c', (2, 1))]\n"
     ]
    }
   ],
   "source": [
    "# join\n",
    "rdd5 = sc.parallelize(['a','b','c'])\n",
    "rdd6 = sc.parallelize(['b','c','d'])\n",
    "rdd7 = rdd5.zipWithIndex()\n",
    "rdd8 = rdd6.zipWithIndex()\n",
    "rdd8 = rdd7.join(rdd8)\n",
    "print(rdd8.collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "96289b1f",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('hello', 1), ('hello', 1), ('hello', 1), ('pyspark', 1), ('python', 1), ('spark', 1)]\n"
     ]
    }
   ],
   "source": [
    "# sortByKey\n",
    "rdd8 = rdd1.sortByKey()\n",
    "print(rdd8.collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "a193327b",
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "defaultdict(<class 'int'>, {'hello': 3, 'spark': 1, 'python': 1, 'pyspark': 1})\n"
     ]
    }
   ],
   "source": [
    "# countByKey\n",
    "result = rdd1.countByKey()\n",
    "print(result)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "c9d20b5e",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(52, 11) 0.21153846153846154\n"
     ]
    }
   ],
   "source": [
    "# aggregate 平均  seqFunc, combFunc\n",
    "rdd9 = sc.parallelize([1,3,6,4,3,5,7,8,4,5,6])\n",
    "result = rdd9.aggregate((0, 0),lambda x, y: (x[0]+y, x[1]+1),lambda m, n: (m[0]+n[0], m[1]+n[1]))\n",
    "print(result, result[1]/result[0])\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "e3e60917",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[(1, 7), (2, 4)]\n"
     ]
    }
   ],
   "source": [
    "# aggregateByKey 求每个键最大值\n",
    "rdd = sc.parallelize([(1,1),(1,2),(2,1),(2,3),(2,4),(1,7)])\n",
    "\n",
    "def seqFunc(a,b):\n",
    "    return max(a,b)\n",
    "def combFunc(a,b):\n",
    "    return max(a,b)\n",
    "\n",
    "result = rdd.aggregateByKey(1, seqFunc, combFunc)\n",
    "print(result.collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "a759f342",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('B', (4, 2)), ('C', (11, 2)), ('A', (3, 2))]\n"
     ]
    }
   ],
   "source": [
    "# combineByKey  createCombiner, mergeValues, mergeCombiners\n",
    "rdd = sc.parallelize([('A',1),('A',2),('B',1),('B',3),('C',4),('C',7)])\n",
    "result = rdd.combineByKey( \n",
    "    lambda v: (v,1),\n",
    "    lambda acc, v: (acc[0]+value, acc[1]+1),\n",
    "    lambda acc1,acc2: (acc1[0]+acc2[0], acc1[1]+acc2[1])\n",
    ")\n",
    "print(result.collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "6fbfef81",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('Wilma', 95.33333333333333), ('Fred', 91.33333333333333)]\n"
     ]
    }
   ],
   "source": [
    "#练习题1 各自的平均成绩\n",
    "#Fred 和 william 二个人 数语外分数 分别是\n",
    "#val scores = Array((\"Fred\", 88), (\"Fred\", 95), (\"Fred\", 91), (\"Wilma\", 93), (\"Wilma\", 95), (\"Wilma\", 98))\n",
    "rdd = sc.parallelize([(\"Fred\", 88), (\"Fred\", 95), (\"Fred\", 91), (\"Wilma\", 93), (\"Wilma\", 95), (\"Wilma\", 98)])\n",
    "result = rdd.aggregateByKey(\n",
    "    (0,0),\n",
    "    lambda x, y: (x[0]+y, x[1]+1),\n",
    "    lambda m, n: (m[0]+n[0], m[1]+n[1])\n",
    ").map(lambda x: (x[0], x[1][0] / x[1][1]))\n",
    "print(result.collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "a761ace4",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('5', 33), ('4', 22), ('6', 39), ('3', 18), ('8', 32)]\n",
      "['6月', '5月']\n"
     ]
    }
   ],
   "source": [
    "# 练习题2 统计每一个月最高温度的两天\n",
    "# 原始数据:\n",
    "# 2019-6-1\t39\n",
    "# 2019-5-21\t33\n",
    "# 2019-6-1\t38\n",
    "# 2019-6-2\t31\n",
    "# 2018-3-11\t18\n",
    "# 2018-4-23\t22\n",
    "# 1970-8-23\t23\n",
    "# 1970-8-8\t32\n",
    "rdd = sc.parallelize([\"2019-6-1\t39\",\"2019-5-21\t33\",\"2019-6-1\t38\",\"2019-6-2\t31\",\"2018-3-11\t18\",\"2018-4-23\t22\",\"1970-8-23\t23\",\"1970-8-8\t32\"])\n",
    "\n",
    "def msplit(line):\n",
    "    t = line.split(\"\\t\")\n",
    "    return (t[0].split(\"-\")[1], int(t[1]))\n",
    "\n",
    "rdd1 = rdd.map(msplit)\n",
    "rdd2 = rdd1.reduceByKey(lambda x, y: max(x,y))\n",
    "print(rdd2.collect())\n",
    "rdd3 = rdd2.sortBy(lambda x: x[1], ascending=False)\n",
    "print(rdd3.map(lambda x: x[0]+'月').take(2))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b7c984e1",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 练习题3 从Nginx服务器日志中获取每个时间段访问量\n",
    "# 139.207.57.81 - - [12/Oct/2022:07:34:50 +0000] \"请求方法和路径...\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2299fd2d",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 练习题4 统计Nginx日志中获取2022年8月每日的访问量\n",
    "# 139.207.57.81 - - [12/Oct/2022:07:34:50 +0000] \"请求方法和路径...\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6acb7051",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 练习题5 访问记录分析\n",
    "# http://math.lynu.edu.cn/zhangsan 其中math表示科目, 张三表示教师\n",
    "# 编写Spark程序统计下面的问题:\n",
    "# (1)各个科目所有教师访问量Top5的\n",
    "# (2)每个科目访问量Top3的教师\n",
    "# (3)访问量最高的5个科目\n",
    "# (4)访问量最高的5个教师"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "60d46d50",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 练习题6 成绩分析\n",
    "# 有文本文件数据源格式如下:(姓名,科目,分数)\n",
    "# Tom,Database,80\n",
    "# Jim,Database,90\n",
    "# Tom,DataStructure,80\n",
    "# Jim,DataStructure,97\n",
    "# ...\n",
    "# 编程实现以下统计结果\n",
    "# (1)总共有多少名学生\n",
    "# (2)该学校开始了多少们课程\n",
    "# (3)Tom总成绩平均分\n",
    "# (4)每名同学选修的课程门数\n",
    "# (5)DataStructure课程有多少人选修\n",
    "# (6)各门课程的平均分是多少\n",
    "# (7)每位学生的总成绩"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c3c132f2",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 练习题7 计算每个学生分数最高的3个成绩 \n",
    "# Andy,99\n",
    "# Mike,88\n",
    "# Bill,99\n",
    "# Bill,88,\n",
    "# Andy,87"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "133aca02",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 练习题8 计算每日用户新增数量\n",
    "# 2022-10-01 user1\n",
    "# 2022-10-01 user2\n",
    "# 2022-10-01 user3\n",
    "# 2022-10-02 user1\n",
    "# 2022-10-02 user2\n",
    "# 2022-10-02 user4\n",
    "# 2022-10-03 user2\n",
    "# 2022-10-03 user5\n",
    "# 2022-10-03 user6"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "05264536",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
